﻿using System;
using System.Collections.Generic;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.ServiceModel.Description;
using System.Threading;

namespace XchangeStreamer.Server
{
    [ServiceBehavior(ConcurrencyMode = ConcurrencyMode.Multiple, InstanceContextMode = InstanceContextMode.Single)]
    public class Streamer : IPublisher, ISubscriber, IDisposable
    {
        #region Private Fields

        private int topicId = 0;
        private ReaderWriterLockSlim syncTopics = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
        private ReaderWriterLockSlim syncSubscribers = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
        private List<Topic> topics = new List<Topic>();
        private Dictionary<int, List<Subscriber>> subscribers = new Dictionary<int, List<Subscriber>>();
        private TypedMessageConverter messageConverter = TypedMessageConverter.Create(typeof(XchangeMessage), "Notify");
        private AsyncCallback onNotifyCompleted;
        
        #endregion

        #region Public Const.

        public Streamer()
        {
            this.onNotifyCompleted = new AsyncCallback(NotifyCompleted);
        }

        #endregion

        #region IPublisher Members

        public void Publish(XchangeMessage message) 
        {
            Console.WriteLine("Time: {0} | MessageId: {1}", message.Time, message.MessageId);
            this.syncSubscribers.EnterReadLock();
            List<Subscriber> subs;
            var res = this.subscribers.TryGetValue(message.TopicId, out subs);
            if (res && subs.Count > 0)
            {
                var msg = messageConverter.ToMessage(message);
                MessageBuffer notificationMessageBuffer = msg.CreateBufferedCopy(65536);
                msg.Close();

                foreach (var sub in subs)
                {
                    sub.Callback.BeginNotify(notificationMessageBuffer.CreateMessage(), this.onNotifyCompleted, sub.Callback);
                }
                notificationMessageBuffer.Close();
            }
            this.syncSubscribers.ExitReadLock();

            if (!res)
            {
                throw new InvalidTopicIdException("");
            }
        }

        public int CreateTopic(string topicName)
        {
            int topId = Interlocked.Increment(ref this.topicId);
            this.syncTopics.EnterWriteLock();
            this.topics.Add(new Topic(topId, topicName));
            this.syncSubscribers.EnterWriteLock();
            this.subscribers.Add(topId, new List<Subscriber>());
            this.syncSubscribers.ExitWriteLock();
            this.syncTopics.ExitWriteLock();
            return topId;
        }

        public void DeleteTopic(int topicId)
        {
            this.syncTopics.EnterWriteLock();
            int res = this.topics.RemoveAll(t => t.Id == topicId);
            if (res > 0)
            {
                this.syncSubscribers.EnterWriteLock();
                this.subscribers.Remove(topicId);
                this.syncSubscribers.ExitWriteLock();
            }
            this.syncTopics.ExitWriteLock();

            if (res == 0)
            {
                throw new InvalidTopicIdException("");
            }
        }

        public void UpdateTopic(int topicId, string topicName)
        {
            this.syncTopics.EnterReadLock();
            var res = this.topics.Find(t => t.Id == topicId);
            if (res != null)
            {
                res.Name = topicName;
            }
            this.syncTopics.ExitReadLock();

            if (res == null)
            {
                throw new InvalidTopicIdException("");
            }
        }

        #endregion

        #region Private Methods

        private void NotifyCompleted(IAsyncResult result)
        {
            ((INotification)(result.AsyncState)).EndNotify(result);
        }

        private void Unsubscribe(object sender, EventArgs e)
        {
            IContextChannel channel = (IContextChannel)sender;
            if (channel != null)
            {
                channel.Faulted -= new EventHandler(this.Unsubscribe);
                channel.Closed -= new EventHandler(this.Unsubscribe);

                if (channel.SessionId != null)
                {
                    this.syncSubscribers.EnterUpgradeableReadLock();
                    Subscriber res = null;
                    foreach (var topic in this.subscribers)
                    {
                        foreach (var sub in topic.Value)
                        {
                            if (sub.SessionId == channel.SessionId)
                            {
                                res = sub;
                                break;
                            }
                        }
                        if (res != null)
                        {
                            this.syncSubscribers.EnterWriteLock();
                            topic.Value.Remove(res);
                            this.syncSubscribers.ExitWriteLock();
                            res = null;
                            break;
                        }
                    }
                    this.syncSubscribers.ExitUpgradeableReadLock();
                }
            }
        }

        #endregion

        #region ISubscriber Members

        public void Subscribe(int topicId)
        {
            var subscriber = new Subscriber(OperationContext.Current.Channel.SessionId, OperationContext.Current.GetCallbackChannel<INotification>());

            List<Subscriber> subs;
            this.syncSubscribers.EnterUpgradeableReadLock();
            var res = this.subscribers.TryGetValue(topicId, out subs);
            if (res)
            {
                if (subs.Find(s => s.SessionId == subscriber.SessionId && s.Callback == subscriber.Callback) == null)
                {
                    this.syncSubscribers.EnterWriteLock();
                    subs.Add(subscriber);
                    this.syncSubscribers.ExitWriteLock();
                    ((IContextChannel)subscriber.Callback).Faulted += new EventHandler(this.Unsubscribe);
                    ((IContextChannel)subscriber.Callback).Closed += new EventHandler(this.Unsubscribe);
                }
            }
            this.syncSubscribers.ExitUpgradeableReadLock();

            if (!res)
            {
                throw new InvalidTopicIdException("");
            }
        }

        public void Unsubscribe(int topicId)
        {
            var callback = OperationContext.Current.GetCallbackChannel<INotification>();

            List<Subscriber> subs;
            this.syncSubscribers.EnterUpgradeableReadLock();
            var res = this.subscribers.TryGetValue(topicId, out subs);
            if (res)
            {
                var sub = subs.Find(s => s.Callback == callback);
                if (sub != null)
                {
                    ((IContextChannel)sub.Callback).Faulted -= new EventHandler(this.Unsubscribe);
                    ((IContextChannel)sub.Callback).Closed -= new EventHandler(this.Unsubscribe);
                    this.syncSubscribers.EnterWriteLock();
                    subs.Remove(sub);
                    this.syncSubscribers.ExitWriteLock();
                }
            }
            this.syncSubscribers.ExitUpgradeableReadLock();

            if (!res)
            {
                throw new InvalidTopicIdException("");
            }
        }

        #endregion

        #region IDisposable Members

        public void Dispose()
        {

        }

        #endregion

        #region Nested Types

        private class Topic
        {
            private Topic()
            { }

            public Topic(int id, string name)
            {
                this.Id = id;
                this.Name = name;
            }

            public int Id
            {
                get;
                set;
            }

            public string Name
            {
                get;
                set;
            }
        }

        private class Subscriber
        {
            private Subscriber()
            { }

            public Subscriber(string sessionId, INotification callback)
            {
                this.SessionId = sessionId;
                this.Callback = callback;
            }

            public string SessionId
            {
                get;
                set;
            }

            public INotification Callback
            {
                get;
                set;
            }
        }

        #endregion
    }
}

